package com.cn.daimajiangxin.flink;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Arrays;

public class StreamingWordCount {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 明确设置为流处理模式
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

        // 启用检查点
        env.enableCheckpointing(5000);

        // 创建Kafka源（无界数据源）
        KafkaSource<String> source = KafkaSource.<String>
                        builder()
                .setBootstrapServers("localhost:9092")
                .setTopics("word-count-topic")
                .setGroupId("flink-group")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        // 从Kafka读取数据
        DataStream<String> text = env.fromSource(
                source,
                WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(2)),
                "Kafka Source"
        );

        // 数据处理逻辑
        DataStream<Tuple2<String, Integer>> counts = text
                .flatMap(new Tokenizer())
                .keyBy(value -> value.f0)
                .sum(1);

        // 输出结果
        counts.print();

        // 执行作业
        env.execute("Streaming Word Count");
    }

    public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
        private static final long serialVersionUID = 1L;
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
            Arrays.stream(value.toLowerCase().split("\\W+"))
                    .filter(word -> word.length() > 0)
                    .forEach(word -> out.collect(new Tuple2<>(word, 1)));
        }
    }
}